package com.youbu.bigdata.utils;

import com.youbu.bigdata.bean.SparkOptionConfig;
import org.apache.commons.lang.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.util.HashMap;
import java.util.Map;

/**
 * Saprk工具类
 * @author sunfangwei
 **/
public class SparkUtil {

	/**
	 * 获取SparkSession
	 * @return
	 */
	public static SparkSession getSparkSession(){
		SparkSession sparkSession = SparkSession.builder().config(getDefaultSparkConfig()).getOrCreate();
		return sparkSession;
	}

	/**
	 * 获取SparkSession
	 * @return
	 */
	public static SparkSession getSparkSession(SparkConf conf){
		SparkSession sparkSession = SparkSession.builder().config(conf).getOrCreate();
		return sparkSession;
	}


	/**
	 * 获取配置
	 * @return
	 */
	private static SparkConf getDefaultSparkConfig(){
		SparkConf conf = new SparkConf();
		Map<String,String> confParam = (Map<String, String>) SparkConfigPropertiesUtil.getAll();
		if(confParam.size()>0){
			for(String key:confParam.keySet()){
				String value = confParam.get(key);
				if(StringUtils.isNotBlank(value)){
					conf.set(key,value);
				}
			}
			return conf;
		}
		return null;
	}

	/**
	 * 保存数据
	 * @param dataset 结果数据
	 * @param optionConfig 参数设置
	 */
	public static void jdbcSaveData(Dataset<?> dataset, SparkOptionConfig optionConfig){
		dataset.write().format("jdbc").options(optionConfig.getOptionMap()).save();
	}

	/**
	 * 获取JDBC数据
	 * @param sparkSession
	 * @param optionConfig
	 * @return
	 */
	public static Dataset<Row> getJDBCDataSet(SparkSession sparkSession,  SparkOptionConfig optionConfig){
		return sparkSession.read().format("jdbc").options(optionConfig.getOptionMap()).load().cache();
	}

	/**
	 * 获取文件内容
	 * @param sparkSession
	 * @param path
	 * @return
	 */
	public static JavaRDD<String>  getTexteFile(SparkSession sparkSession,String path){
		return sparkSession.read().textFile(path).javaRDD();
	}


	public static Map<String,String> getOptins(){
		Map<String,String> map = new HashMap<>();
		return map;
	}




}
